package com.wasp.collector.service.akka;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import com.wasp.collector.data.TimeInterval;
import com.wasp.collector.service.akka.StatsMessages.JobFailed;
import com.wasp.collector.service.akka.StatsMessages.StatsJob;
import com.wasp.collector.service.akka.StatsMessages.StatsResult;
import com.wasp.collector.service.akka.cluster.AkkaService;

import akka.actor.AbstractActor;
import akka.actor.ActorSelection;
import akka.actor.Address;
import akka.japi.pf.ReceiveBuilder;
import scala.PartialFunction;
import scala.concurrent.duration.Duration;
import scala.concurrent.duration.FiniteDuration;
import scala.runtime.BoxedUnit;

public class CollectActorClient extends AbstractActor {
	final String servicePath = AkkaService.getServicePath();
	final INotify notify;
	public CollectActorClient(INotify notify) {
		this.notify = notify;
		FiniteDuration interval = Duration.create(2, TimeUnit.SECONDS);
	}

	@Override
	public PartialFunction<Object, BoxedUnit> receive() {
		return ReceiveBuilder.match(HttpClientActorContext.class, t -> {
			sendMsg(t);
		})
		.match(TimeInterval.class, t->{
			sendMsg(t);
		})
		.match(ActorResult.class, t->{
			notify.completed(t);
			stopSelf();
		})
		.match(JobFailed.class, System.out::println)
		.build();
	}
	
	private void sendMsg(Object obj){
		Address address = AkkaService.balance();
		ActorSelection service = getContext().actorSelection(address + servicePath);
		service.tell(obj, self());
	}
	private void stopSelf(){
		getContext().stop(self());
	}

}
